package com.navinfo.tripanalysis.offline;

import com.navinfo.location.mileage.bean.MileageAndFuel;
import com.navinfo.platform.score.TripScoreService;
import com.navinfo.platform.score.pojo.ScoreWeight;
import com.navinfo.tripanalysis.common.arithmetic.common.*;
import com.navinfo.tripanalysis.common.pojo.*;
import com.navinfo.tripanalysis.common.util.*;
import com.navinfo.tripanalysis.common.vehicle.VehicleInfo;
import com.navinfo.tripanalysis.offline.pojo.AlgorithmChainCommand;
import com.navinfo.tripanalysis.offline.pojo.AlgorithmOuterData;
import com.navinfo.tripanalysis.offline.pojo.LoadDataParam;
import com.navinfo.tripanalysis.offline.pojo.PreTripInfo;
import com.navinfo.tripanalysis.offline.service.*;
import com.navinfo.tripanalysis.offline.service.impl.LoadCn6ThresholdDataInfoServiceImpl;
import com.navinfo.tripanalysis.offline.service.impl.huashen.LoadCn6BasicDataInfoServiceImpl;
import com.navinfo.tripanalysis.offline.util.TripStatisticFixUtils;
import com.navinfo.tripanalysis.offline.util.TripStatisticUtils;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.Options;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.Optional;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.storage.StorageLevel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.Tuple2;

import java.util.*;
import java.util.stream.Collectors;

import static com.navinfo.tripanalysis.offline.util.SparkAppUtils.getSparkBuilder;

/**
 * 驾驶行为分析离线模块<br/>
 * 通过公共轨迹协议点来计算
 * 华神使用 为主入口
 * @author 王恩宝
 */
public class TripAnalysisNewApplication extends TripAnalysisApplication{
    private static final Logger logger = LoggerFactory.getLogger(TripAnalysisNewApplication.class);

    public static void main(String[] args) throws Exception {
        long startTime = System.currentTimeMillis();
        logger.error("开始离线驾驶行为分析，args:{}", Arrays.toString(args));

        //加载配置
        Options options = setCmdOptions();
        CommandLine cmd = parseCmdOptions(options, args);
        Properties config = loadProps(cmd);
        parseDays(args, cmd);

        String runType = config.getProperty("system.running.type");
        if(StringUtils.isEmpty(runType)){
            logger.error("请配置系统运行类型，1东风，2青汽");
            System.exit(1);
        }

        runningTypeEnum = RunningTypeEnum.valueOf(Integer.parseInt(runType));
        logger.error("离线驾驶行为分析，系统运行类型：{},{}", runningTypeEnum.getType(), runningTypeEnum.getDesc());


        //加载回放特定tid的列表
        LoadDataParam param = new LoadDataParam();
        param.setTidList(loadTidList(cmd));

        //初始化SparkSession
        SparkSession spark = getSparkBuilder(config).getOrCreate();
        JavaSparkContext jsc = new JavaSparkContext(spark.sparkContext());

        //创建相关服务
        ServiceFactory factory = new ServiceFactory(config, runningTypeEnum);
        DataBatchService dataBatchService = factory.createDataBatchService();
        LoadPointProtocolService loadPointProtocolService = factory.createLoadPointService();
        DrivingDataService drivingDataService = factory.createDrivingDataService();
        List<SaveTripStatisticService> saveTripStatisticServices = factory.createSaveTripServices();
        List<SaveTripEventService> saveTripEventServices = factory.createSaveEventServices();
        TripStatisticExtendService tripStatisticExtendService = factory.createTripStatisticExtendService();

        //新增发动机数据---------------------------------------------------------------------------------------------------
        //国六车辆基础信息
        LoadCn6BasicDataInfoServiceImpl cn6Service = factory.createLoadCn6BasicDataService();
        PointSortGroupService sortGroupService = factory.createPointSortGroupService();
        //创建发动机流相关服务
        List<SaveTripEngineDataService> saveTripEngineDataServices = factory.createSaveTripEngineDataServices();
        LoadEngineDataService loadEngineService = factory.createLoadEngineDataService();
//        DataCombineService dataCombineService = factory.createDataCombineService();

        //加载cn6需求车辆基础信息
        Broadcast<HashMap<Long, Cn6BasicDataInfo>> cn6BasicDataInfoBroadcast = cn6Service.getCn6BasicDataInfoRdd(spark, jsc);
//        //加载发动机数据流异常阈值
        Broadcast<HashMap<String,Cn6EngineDataThresholdInfo>> cn6ThresholdInfoBroadcast = cn6EngineDataThresholdInfoBroadcast(spark, jsc,factory);
        //新增发动机数据---------------------------------------------------------------------------------------------------

        //是否加载瓦片数据
        boolean loadTileData = Boolean.parseBoolean(config.getProperty("load.data.tile.open"));
        logger.error("loadTileData:{}", loadTileData);
        //广播变量：瓦片与区域信息、车辆配置信息、算法配置信息
//        final Broadcast<Map<Long, Integer>> tileLcBroadcast = jsc.broadcast(loadTileData ? new HashMap<>(factory.createLoadTileService().load(spark, jsc).collectAsMap()) : new HashMap<>() );
        final Broadcast<Map<Long, Integer>> tileLcBroadcast = null;
        final Broadcast<Map<Long, VehicleInfo>> vehicleInfoBroadcast = vehicleInfoBroadcast(spark, jsc, factory );
        final Broadcast<Map> algorithmConfBroadcast = algorithmConfBroadcast(jsc, config);

        //添加行程打分算法与权重信息
        Map<String, ScoreWeight> score = new HashMap<>(TripScoreService.getLoadAndEnvironmentData());
        final Broadcast<Map<String, ScoreWeight>> scoreBroadcast = jsc.broadcast(score);

        //广播变量：系统运行类型
        final Broadcast<RunningTypeEnum> runTypeBroadcast = jsc.broadcast(runningTypeEnum);

        //按天进行行程和事件的计算
        for(long sTime=start; sTime<end; sTime+= DAY_MS){
            Date currentDay = new Date(sTime);
            param.setDay(currentDay);

            //获取昨天的中间变量
            final Broadcast<Map<Long, DrivingData>> yesterdayDrivingData = drivingDataBroadcast(spark, jsc, drivingDataService, sTime);

            //最终的输出结果
            List<JavaPairRDD<Long, AlgorithmOuterData>> outDataRDDs = new ArrayList<>();

            //按批次处理数据
            List<LoadDataParam> batchParams = dataBatchService.createBatch(param);
            for (LoadDataParam oneBatchParam : batchParams) {
                logger.error("========>oneBatchParam:{}", oneBatchParam);

                //新增发动机数据-------------------------------------------------------------------------------------------
                //加载发动机的数据
                JavaRDD<Point> pointsEngine = loadEngineService.load(spark, jsc, oneBatchParam);
                if (pointsEngine == null) {
                    logger.error("OOO:engine data is null");
                    pointsEngine = jsc.parallelize(new ArrayList<>());
                }
                logger.error("OOO:engine data size:{}", pointsEngine.count());
                //分组排序发动机数据
                JavaPairRDD<Long, List<Point>> tidPointsEngine = sortGroupService.sortAndGroup(pointsEngine);
                //新增发动机数据-------------------------------------------------------------------------------------------

                //加载公共轨迹协议的数据
                JavaPairRDD<Long, AlgorithmOuterData> outDataRDD = (JavaPairRDD<Long, AlgorithmOuterData>) loadPointProtocolService.load(spark, jsc, oneBatchParam)
                        .mapToPair((PairFunction<Tuple2<Long, List<PointProtocol>>, Long, List<PointProtocol>>) tuple2 -> new Tuple2<>(tuple2._1, tuple2._2))
                        .leftOuterJoin(tidPointsEngine)
                        .mapPartitionsToPair((PairFlatMapFunction<Iterator<Tuple2<Long, Tuple2<List<PointProtocol>, Optional<List<Point>>>>>, Long, AlgorithmOuterData>) it ->
                                new Iterator<Tuple2<Long, AlgorithmOuterData>>() {
                                    final AlgorithmChainService algorithmChain = factory.createAlgorithmChainService(runTypeBroadcast.getValue(),algorithmConfBroadcast.getValue());
                                    final PreTripInfoService preTripService = ServiceFactory.createServiceImpl(runTypeBroadcast.getValue(), PreTripInfoService.class );

                                    @Override
                                    public boolean hasNext() {
                                        return it.hasNext();
                                    }

                                    @Override
                                    public Tuple2<Long, AlgorithmOuterData> next() {
                                        Tuple2<Long, Tuple2<List<PointProtocol>, Optional<List<Point>>>> tidPointTuple = it.next();
                                        Long tid = tidPointTuple._1();
                                        Tuple2<List<PointProtocol>, Optional<List<Point>>> pointTuple2 = tidPointTuple._2();
                                        List<PointProtocol> points = pointTuple2._1;
                                        Optional<List<Point>> optionalPointsEngine = pointTuple2._2();
                                        List<Point> pointsEngine = new ArrayList<>();
                                        if (optionalPointsEngine.isPresent()) {
                                            pointsEngine = optionalPointsEngine.get();
                                        }
                                        List<Point> points0F37 = new ArrayList<>();
                                        List<Point0200> points0200 = new ArrayList<>();

                                        points.forEach(e -> {
                                            if(ProtocolEnum.LOCATION.equals(e.getProtocol())){
                                                points0200.add(PointUtils.toPoint0200(e));
                                            }else if(ProtocolEnum.REALTIME.equals(e.getProtocol())){
                                                points0F37.add(e);
                                            }
                                        });

                                        //如果只根据0f37来生成行程开关，当前没有0f37点，则直接返回一个空对像
                                        if (Boolean.parseBoolean(config.getProperty("only0F37Trip")) && CommonUtils.isCollectionEmpty(points0F37)) {
                                            return new Tuple2<>(tid, new AlgorithmOuterData());
                                        }

                                        //合并车辆的0f37流和发动机流
                                        Map<Long, Cn6BasicDataInfo> basicDataInfoMap = cn6BasicDataInfoBroadcast.getValue();
                                        List<Point> combinePoints = new ArrayList<>();
                                        Cn6BasicDataInfo cn6BasicDataInfo = basicDataInfoMap.get(tid);
                                        String engineModelCode = "";
                                        if (CommonUtils.isCollectionNotEmpty(points0F37) && cn6BasicDataInfo != null && StringUtils.isNotBlank(cn6BasicDataInfo.getVin())) {
                                            engineModelCode = cn6BasicDataInfo.getEngineModelCode();
                                            //获取0F37最大的时间点
                                            Long maxGpstime = points0F37.stream().max((o1, o2) -> (int) (o1.getGpsTime() - o2.getGpsTime())).map(Point::getGpsTime).orElse(0L);
                                            //国六数据流时间在0f37之后的都滤掉
                                            pointsEngine = pointsEngine.stream().filter(e -> e.getGpsTime() < maxGpstime).collect(Collectors.toList());
                                            points0F37.addAll(pointsEngine);
                                            combinePoints = points0F37.stream().sorted((p1, p2) ->  (int) (p1.getGpsTime() - p2.getGpsTime())).collect(Collectors.toList());

                                            int lastTripId = 1;
                                            for (Point point : combinePoints) {
                                                if (point.getProtocol().equals(ProtocolEnum.ENGINE)) {
                                                    ((PointEngine) point).setTripId(lastTripId);
                                                } else {
                                                    lastTripId = point.getTripId();
                                                }
                                            }
                                        }


                                        // 获取对应车辆的发动机阈值
                                        HashMap<String,Cn6EngineDataThresholdInfo> cn6EngineDataThresholdInfoHashMap = cn6ThresholdInfoBroadcast.getValue();
                                        Cn6EngineDataThresholdInfo cn6EngineDataThresholdInfo = new Cn6EngineDataThresholdInfo();
                                        if(StringUtils.isNotBlank(engineModelCode)){
                                            cn6EngineDataThresholdInfo = cn6EngineDataThresholdInfoHashMap.get(engineModelCode);
                                            if(null == cn6EngineDataThresholdInfo){
                                                cn6EngineDataThresholdInfo = new Cn6EngineDataThresholdInfo();
                                            }
                                        }

                                        //构建算法链的输出对象
                                        AlgorithmOuterData outerData = new AlgorithmOuterData();
                                        outerData.setOutStatistics(new ArrayList<>());
                                        outerData.setOutEngineStatistics(new ArrayList<>());
                                        outerData.setOutEvents(CommonUtils.isCollectionNotEmpty(saveTripEventServices) ? new ArrayList<>() : null);

                                        //初始化中间数据
                                        DrivingData drivingData = yesterdayDrivingData.getValue().get(tid);
                                        if (drivingData == null) {
                                            drivingData = new DrivingData();
                                            drivingData.getCommonData().setTid(tid);
                                        }

                                        drivingData.setVehicleInfo(vehicleInfoBroadcast.getValue().get(tid));
                                        drivingData.setTileLcBroadcast(tileLcBroadcast);
                                        drivingData.setCn6BasicDataInfo(cn6BasicDataInfo);
                                        drivingData.setEngineDataThresholdInfo(cn6EngineDataThresholdInfo);
                                        logger.error("drivingData:{}",drivingData);

                                        //算法链执行的命令对象
                                        AlgorithmChainCommand command = new AlgorithmChainCommand();
                                        command.setTid(tid);
                                        command.setDay(currentDay);
                                        command.setOuterData(outerData);
                                        command.setDrivingData(drivingData);

                               /*行程切分具体流程<br/>
                                1）根据当天公共轨迹协议点，按tripId切分出子行程，加入对应的0f37及0200的子数据，设置行程开始、结束相关信息；<br/>
                                2) 根据0200的子行程调用精准里程油耗算法算出子行程的精准里程、精准油耗；<br/>
                                3）根据子行程的0200、0F37数据，调用算法链，算出对应的积分里程、积分油耗、积分时长及算法的单项指标值；<br/>
                                4）根据当天的0200点，调用精准里程油耗算法算出当天总的精准里程油耗值；<br/>
                                5）根据子行程的精准里程(油耗)占所有行程的总精准里程(油耗)比例 * 当天总的精准里程(油耗)，算出每个子行程的精准里程（油耗）<br/>
                                6）使用6相同的算法，计算出行程中其他指标的精准值；<br/>
                                */
                                        List<PreTripInfo> tripList = preTripService.judgeTrip(combinePoints, points0200, currentDay);
                                        if (CommonUtils.isCollectionNotEmpty(tripList) ) {
                                            int tripSize = tripList.size();
                                            if (logger.isInfoEnabled()) {
                                                logger.info("tid:{},day:{},tripSize:{}",tid, currentDay, tripSize );
                                            }

                                            //步骤3切分行程的精准里程和油耗数组
                                            Integer[] tripFuel = new Integer[tripSize];
                                            Integer[] tripMileage = new Integer[tripSize];

                                            for (int i=0; i<tripSize; i++) {
                                                PreTripInfo preTripInfo = tripList.get(i);
                                                tripFuel[i] = preTripInfo.getPreciseFuel();
                                                tripMileage[i] = preTripInfo.getPreciseMileage();

                                                command.setPoints0f37(preTripInfo.getSubPoints0f37());
                                                command.setPoints0200(preTripInfo.getSubPoints0200());
                                                if (i == (tripSize-1))  {
                                                    command.setHandleCrossDay(true);
                                                } else {
                                                    command.setHandleCrossDay(false);
                                                }

                                                if (logger.isInfoEnabled()) {
                                                    logger.info("--->i={}", i);
                                                    logger.info("----{}\n", preTripInfo);
                                                    long lastGpsTime = drivingData.getCommonData().getLastGpsTime();
                                                    if (lastGpsTime <= 0) {
                                                        logger.info("--->lastGpsTime is null");
                                                    } else {
                                                        logger.info("--->lastGpsTime: {}", DateUtils.format(new Date(lastGpsTime), DateUtils.DateFormat.YYYY_MM_DD_HH_MM_SS));
                                                        if (lastGpsTime > preTripInfo.getStartTime()) {
                                                            logger.error("Error:lastGpsTime>info.getStartTimeMs");
                                                        }
                                                    }
                                                }

                                                //4、执行算法链，算出行程对应的积分里程、积分油耗、积分时长及算法的单项指标值
                                                algorithmChain.execute(command);
                                            }

                                            //5、根据0200计算出当天总的精准里程油耗值
                                            MileageAndFuel mileageAndFuel = TripStatisticUtils.calcMileageFuel(runTypeBroadcast.getValue() ,points0200);
                                            int dayMileage = ArithmeticUtils.mileageKMToM(mileageAndFuel.getStdMileage());
                                            int dayFuel = ArithmeticUtils.fuelLToMl(mileageAndFuel.getStdFuelCon());

                                            //6、根据子行程的精准里程(油耗)占所有行程的总精准里程(油耗)比例 * 当天总的精准里程(油耗)，算出每个子行程的精准里程（油耗）
                                            List<OuterStatisticData> outTrips = outerData.getOutStatistics();
                                            List<EngineAggregatedData> outEngineDatas = outerData.getOutEngineStatistics();
                                            if (outTrips.size() == tripSize) {
                                                //精准后各个行程的里程油耗
                                                Integer[] correctMileage = TripStatisticFixUtils.fixArrayData(dayMileage, tripMileage);
                                                Integer[] correctFuel = TripStatisticFixUtils.fixArrayData(dayFuel, tripFuel);
                                                for (int i=0; i<tripSize; i++) {
                                                    PreTripInfo preTripInfo = tripList.get(i);
                                                    OuterStatisticData statisticData = outTrips.get(i);
                                                    EngineAggregatedData engineAggregatedData = outEngineDatas.get(i);

                                                    statisticData.setTripId(preTripInfo.getTripId());
                                                    statisticData.setForceEndFlag(preTripInfo.getForceEndFlag());

                                                    //根据0200精准行程的里程油耗
                                                    statisticData.setTripFuel(correctFuel[i]);
                                                    statisticData.setTripMileage(correctMileage[i]);
                                                    statisticData.setPreciseFlag(OuterStatisticData.PRECISE_YES);

                                                    //根据0200精准行程开始、结束的相关信息
                                                    statisticData.setRouteStartTime(preTripInfo.getStartTime() / 1000L);
                                                    statisticData.setRouteStartLatitude(preTripInfo.getStartLatitude());
                                                    statisticData.setRouteStartLongitude(preTripInfo.getStartLongitude());
                                                    statisticData.setRouteStartHeight(preTripInfo.getStartHeight());
                                                    statisticData.setRouteStartDirection(preTripInfo.getStartDirection());
                                                    statisticData.setRouteEndTime(preTripInfo.getEndTime() / 1000L);
                                                    statisticData.setRouteEndLongitude(preTripInfo.getEndLongitude());
                                                    statisticData.setRouteEndLatitude(preTripInfo.getEndLatitude());
                                                    statisticData.setRouteEndHeight(preTripInfo.getEndHeight());
                                                    statisticData.setRouteEndDirection(preTripInfo.getEndDirection());

                                                    statisticData.setRouteStartOilPercentage(preTripInfo.getStartOilPercentage().intValue());
                                                    statisticData.setRouteEndOilPercentage(preTripInfo.getEndOilPercentage().intValue());
                                                    statisticData.setTripDuration((int) (statisticData.getRouteEndTime() - statisticData.getRouteStartTime()));
                                                    if (tileLcBroadcast != null) {
                                                        Map<Long, Integer> value = tileLcBroadcast.getValue();
                                                        statisticData.setRouteStartLc(ArithmeticUtils.getLc(statisticData.getRouteStartLatitude(), statisticData.getRouteStartLongitude(), value));
                                                        statisticData.setRouteEndLc(ArithmeticUtils.getLc(statisticData.getRouteEndLatitude(), statisticData.getRouteEndLongitude(), value));
                                                    }

                                                    //7）使用6相同的算法，计算出行程中其他指标的精准值；
                                                    TripStatisticFixUtils.fixTrips(statisticData);
                                                    //计算行程中发动机数据缺项和异常的精确时长，非国六车辆和国六vin为空的车辆，都不校正精准时长
                                                    if (engineAggregatedData != null) {
                                                        FixEngineAggregatedDataUtils.fixEngineAggregatedData(statisticData, engineAggregatedData);
                                                    }

                                                    //通过0200精准行程开始结束时的里程、油耗
                                                    TripStatisticUtils.calcStartEndMileageFuel(statisticData, preTripInfo.getSubPoints0200());

                                                    //计算行程打分
                                                    TripStatisticUtils.calcTripScore(statisticData, scoreBroadcast);
                                                }
                                            } else {
                                                logger.error("OOO->preTrip tripSize is not equals to trip tripSize tid:{},pre_size:{},trip_size:{}", tid, tripSize, outTrips.size());
                                            }
                                        }

                                        //清空drivingdata上无用的数据
                                        DrivingData outDrivingData = outerData.getDrivingData();
                                        if (outDrivingData != null) {
                                            outDrivingData.setVehicleInfo(null);
                                            outDrivingData.setCn6BasicDataInfo(null);
                                            outDrivingData.setEngineAggregatedData(null);
                                            outDrivingData.setEngineDataThresholdInfo(null);
                                            outDrivingData.setTileLcBroadcast(null);
                                            outDrivingData.setUpdateDay(DateUtils.format(currentDay, DateUtils.DateFormat.YYYYMMDD));
                                        }

                                        logger.info("tid:{},day:{},tripSize:{},eventSize:{}",tid, currentDay, outerData.getOutStatistics().size(),  outerData.getOutEvents().size());
                                        return new Tuple2<>(tid, outerData);
                                    }
                                });

                outDataRDDs.add(outDataRDD);
                outDataRDD.persist(StorageLevel.MEMORY_AND_DISK_SER());
                logger.error("========>结束一个批次：oneBatchParam:{}", oneBatchParam);

                //执行算法链
                logger.error("========>当前批次保存的count:" + outDataRDD.count());
            }

            //合并分批处理的数据结果
            JavaPairRDD<Long, AlgorithmOuterData> outData = null;
            if (outDataRDDs.size() > 0) {
                outData = outDataRDDs.get(0);
                for (int i=1; i<outDataRDDs.size(); i++) {
                    JavaPairRDD<Long, AlgorithmOuterData> tmpRdd = outDataRDDs.get(i);
                    if (tmpRdd != null) {
                        outData = outData.union(tmpRdd);
                    }
                }
            }

            //行程、事件、中间数据落盘
            if (outData != null) {
                if (CommonUtils.isCollectionNotEmpty(saveTripStatisticServices)) {
                    logger.error("行程数据落盘...");
                    saveTripData(spark, jsc, saveTripStatisticServices, tripStatisticExtendService, sTime, outData);
                }
                if ( CommonUtils.isCollectionNotEmpty(saveTripEventServices)) {
                    logger.error("事件数据落盘...");
                    saveEventData(spark, jsc, saveTripEventServices, sTime, outData);
                }
                if (CommonUtils.isCollectionNotEmpty(saveTripEngineDataServices)) {
                    logger.error("行程发动机汇总数据落盘...");
                    saveTripEngineData(spark, jsc, saveTripEngineDataServices, sTime, outData);
                }
                if (null != drivingDataService) {
                    logger.error("中间数据落盘...");
                    saveDrivingData(spark, jsc, drivingDataService, sTime, outData, yesterdayDrivingData);
                }
            } else {
                logger.error("无行程数据生成!!!");
            }
        }

        logger.error("完成离线驾驶行为分析，耗时：{}ms", System.currentTimeMillis()-startTime);
        jsc.close();
        System.exit(0);
    }


    /**
     * 广播发动机阈值信息
     */
    static Broadcast<HashMap<String, Cn6EngineDataThresholdInfo>> cn6EngineDataThresholdInfoBroadcast(SparkSession spark, JavaSparkContext jsc, ServiceFactory factory) {
        //加载国六车辆发动机异常数据阈值信息
        LoadCn6ThresholdDataInfoServiceImpl cn6ThresholdService = factory.createLoadCn6ThresholdDataService();
        return jsc.broadcast(new HashMap<>(cn6ThresholdService.getCn6ThresholdInfoRdd(spark,jsc).collectAsMap()));
    }
}
